package org.apache.flink.connector.scylladb.table;

import org.apache.flink.connector.scylladb.sink.ScyllaDBSinkFunction;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.sink.SinkFunctionProvider;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.types.RowKind;

import java.time.Duration;
import java.util.List;
import java.util.Set;
import java.util.HashSet;

/**
 * ScyllaDB 动态表目标实现，支持数据写入操作
 */
public class ScyllaDBDynamicTableSink implements DynamicTableSink {

    private final String url;
    private final String keySpace;
    private final String tableName;
    private final String username;
    private final String password;
    private final int sinkParallelism;
    private final int sinkBufferFlushMaxRows;
    private final Duration sinkBufferFlushInterval;
    private final int sinkMaxRetries;
    private final int connectionPoolLocalSize;
    private final int connectionPoolRemoteSize;
    private final Duration connectionTimeout;
    private final Duration requestTimeout;
    private final String consistencyLevel;
    private final boolean sslEnabled;
    private final String sslKeystorePath;
    private final String sslKeystorePassword;
    private final String sslTruststorePath;
    private final String sslTruststorePassword;
    private final DataType physicalDataType;
    private final Set<String> appendSetFields;
    private final List<String> primaryKeys;

    public ScyllaDBDynamicTableSink(
            String url,
            String keySpace,
            String tableName,
            String username,
            String password,
            int sinkParallelism,
            int sinkBufferFlushMaxRows,
            Duration sinkBufferFlushInterval,
            int sinkMaxRetries,
            int connectionPoolLocalSize,
            int connectionPoolRemoteSize,
            Duration connectionTimeout,
            Duration requestTimeout,
            String consistencyLevel,
            boolean sslEnabled,
            String sslKeystorePath,
            String sslKeystorePassword,
            String sslTruststorePath,
            String sslTruststorePassword,
            DataType physicalDataType,
            Set<String> appendSetFields,
            List<String> primaryKeys) {
        
        this.url = url;
        this.keySpace = keySpace;
        this.tableName = tableName;
        this.username = username;
        this.password = password;
        this.sinkParallelism = sinkParallelism;
        this.sinkBufferFlushMaxRows = sinkBufferFlushMaxRows;
        this.sinkBufferFlushInterval = sinkBufferFlushInterval;
        this.sinkMaxRetries = sinkMaxRetries;
        this.connectionPoolLocalSize = connectionPoolLocalSize;
        this.connectionPoolRemoteSize = connectionPoolRemoteSize;
        this.connectionTimeout = connectionTimeout;
        this.requestTimeout = requestTimeout;
        this.consistencyLevel = consistencyLevel;
        this.sslEnabled = sslEnabled;
        this.sslKeystorePath = sslKeystorePath;
        this.sslKeystorePassword = sslKeystorePassword;
        this.sslTruststorePath = sslTruststorePath;
        this.sslTruststorePassword = sslTruststorePassword;
        this.physicalDataType = physicalDataType;
        this.appendSetFields = appendSetFields != null ? appendSetFields : new HashSet<>();
        this.primaryKeys = primaryKeys != null ? primaryKeys : new java.util.ArrayList<>();
    }

    @Override
    public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
        // ScyllaDB 支持 INSERT 和 UPDATE_AFTER 操作
        // UPDATE_AFTER 在 ScyllaDB 中被视为 UPSERT 操作
        return ChangelogMode.newBuilder()
                .addContainedKind(RowKind.INSERT)
                .addContainedKind(RowKind.UPDATE_AFTER)
                .addContainedKind(RowKind.DELETE)
                .build();
    }

    @Override
    public DynamicTableSink.SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
        final RowType physicalRowType = (RowType) physicalDataType.getLogicalType();
        
        final ScyllaDBSinkFunction sinkFunction = new ScyllaDBSinkFunction(
                url,
                keySpace,
                tableName,
                username,
                password,
                sinkBufferFlushMaxRows,
                sinkBufferFlushInterval,
                sinkMaxRetries,
                connectionPoolLocalSize,
                connectionPoolRemoteSize,
                connectionTimeout,
                requestTimeout,
                consistencyLevel,
                sslEnabled,
                sslKeystorePath,
                sslKeystorePassword,
                sslTruststorePath,
                sslTruststorePassword,
                physicalRowType,
                appendSetFields,
                primaryKeys
        );

        return SinkFunctionProvider.of(sinkFunction, sinkParallelism);
    }

    @Override
    public DynamicTableSink copy() {
        return new ScyllaDBDynamicTableSink(
                url,
                keySpace,
                tableName,
                username,
                password,
                sinkParallelism,
                sinkBufferFlushMaxRows,
                sinkBufferFlushInterval,
                sinkMaxRetries,
                connectionPoolLocalSize,
                connectionPoolRemoteSize,
                connectionTimeout,
                requestTimeout,
                consistencyLevel,
                sslEnabled,
                sslKeystorePath,
                sslKeystorePassword,
                sslTruststorePath,
                sslTruststorePassword,
                physicalDataType,
                appendSetFields,
                primaryKeys
        );
    }

    @Override
    public String asSummaryString() {
        return String.format("ScyllaDB(url=%s, keyspace=%s, table=%s)", 
                url, keySpace, tableName);
    }
} 